package cn.doitedu.olap_agg;

import cn.doitedu.udfs.TimeTrunkFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job01_TrafficOlapAggTable {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql(
                " CREATE TABLE dwd_user_action               (                            "+
                        " 	release_channel STRING,                                       "+
                        " 	device_type STRING,                                           "+
                        " 	session_id STRING,                                            "+
                        " 	event_id STRING,                                              "+
                        " 	action_time BIGINT,                                           "+
                        " 	properties MAP<STRING,STRING>,                                "+
                        "   user_id bigint,                                               "+
                        "   create_time timestamp(3),                                     "+
                        "   province string,                                              "+
                        " 	city string,                                                  "+
                        " 	region string,                       				          "+
                        "   page_type string,                                             "+
                        " 	page_business string,                                         "+
                        " 	page_lanmu string,                                            "+
                        " 	url as properties['url'],                                     "+
                        " 	rt as  to_timestamp_ltz(action_time,3),                       "+
                        " 	watermark for rt as rt                                        "+
                        " ) WITH (                                                        "+
                        "     'connector' = 'kafka',                                      "+
                        "     'topic' = 'dwd-user-action',                                "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',            "+
                        "     'properties.group.id' = 'g008',                             "+
                        "     'scan.startup.mode' = 'latest-offset',                      "+
                        "     'value.format' = 'json',                                    "+
                        "     'value.fields-include' = 'EXCEPT_KEY'                       "+
                        " )				                                                  "
        );


        // sql
        tenv.createTemporaryFunction("time_trunk", TimeTrunkFunction.class);
        tenv.executeSql("create temporary view tmp_res  AS " +
                        "WITH tmp AS (\n" +
                        "    SELECT\n" +
                        "        user_id,\n" +
                        "        session_id,\n" +
                        "        event_id,\n" +
                        "        action_time,\n" +
                        "        province,\n" +
                        "        city,\n" +
                        "        region,\n" +
                        "        device_type,\n" +
                        "        release_channel,\n" +
                        "        url,\n" +
                        "        page_type,\n" +
                        "        page_lanmu,\n" +
                        "        page_business,\n" +
                        "        if(to_date(date_format(create_time,'yyyy-MM-dd'))=current_date,1,0) as old_or_new,\n" +
                        "        rt\n" +
                        "    FROM dwd_user_action\n" +
                        ")\n" +
                        "\n" +
                        "SELECT\n" +
                        "    to_date(date_format(window_start,'yyyy-MM-dd')) as dt, " +
                        "    time_trunk(window_start,60) as time_trunc_1h,          " +
                        "    time_trunk(window_start,30) as time_trunc_hh,          " +
                        "    time_trunk(window_start,10) as time_trunc_10m,         " +
                        "    time_trunk(window_start,5) as time_trunc_5m,           " +
                        "    time_trunk(window_start,1) as time_trunc_1m,           " +
                        "    province,\n" +
                        "    city,\n" +
                        "    region,\n" +
                        "    device_type,\n" +
                        "    release_channel,\n" +
                        "    url,\n" +
                        "    page_type,\n" +
                        "    page_lanmu,\n" +
                        "    page_business,\n" +
                        "    old_or_new,\n" +

                        "    count(event_id) filter(where event_id='page_load') as pv, \n" +
                        "    user_id,       \n" +
                        "    session_id     \n" +

                        "FROM TABLE(\n" +
                        "    TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE)\n" +
                        ")\n" +
                        "GROUP BY \n" +
                        "    window_start,\n" +
                        "    window_end,\n" +
                        "    user_id,\n" +
                        "    session_id,\n" +
                        "    province,\n" +
                        "    city,\n" +
                        "    region,\n" +
                        "    device_type,\n" +
                        "    release_channel,\n" +
                        "    url,\n" +
                        "    page_type,\n" +
                        "    page_lanmu,\n" +
                        "    page_business,\n" +
                        "    old_or_new"
        );


        // 创建doris连接器表，表结构跟上面的查询结果应该完全一致，但是需要在连接器的参数中指定逻辑字段和doris中物理字段的映射关系
        tenv.executeSql(
                        "CREATE TABLE sink_doris(     \n" +
                        "    dt            date,      \n" +
                        "    time_trunc_1h string,    \n" +
                        "    time_trunc_hh string,    \n" +
                        "    time_trunc_10m string,   \n" +
                        "    time_trunc_5m string,    \n" +
                        "    time_trunc_1m string,    \n" +
                        "    province string,\n" +
                        "    city string,\n" +
                        "    region string,\n" +
                        "    device_type string,\n" +
                        "    release_channel string,\n" +
                        "    url string,\n" +
                        "    page_type string,\n" +
                        "    page_lanmu string,\n" +
                        "    page_business string,\n" +
                        "    old_or_new  int ,\n" +
                        "\n" +
                        "    pv  bigint,\n" +
                        "    uv  bigint,  \n" +
                        "    session_cnt  string   \n" +
                        ") WITH (                                         \n" +
                        "   'connector' = 'doris',                        \n" +
                        "   'fenodes' = 'doitedu:8030',                   \n" +
                        "   'table.identifier' = 'dws.tfc_overview_olap_agg',  \n" +
                        "   'username' = 'root',                          \n" +
                        "   'password' = 'root',                          \n" +
                        "   'sink.label-prefix' = 'doris_label-003',      \n" +
                        "   'sink.properties.columns' = 'dt,time_trunc_1h,time_trunc_hh,time_trunc_10m,time_trunc_5m,time_trunc_1m," +
                        "       province,city,region,device_type,release_channel,url,page_type,page_lanmu,page_business," +
                        "       old_or_new,pv,uv,session_cnt,uv=to_bitmap(uv),session_cnt=hll_hash(session_cnt) ' " +
                        ")"
        );


        // sink查询结果到目标表
        tenv.executeSql("insert into sink_doris select * from tmp_res");


    }


}
